package com.guass.nettyprotostuff.service.impl;

import com.guass.nettyprotostuff.conf.ServerProperties;
import com.guass.nettyprotostuff.service.MqService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * @author guass
 * @date 2022年07月06日 18:03
 */
@Service
@Slf4j
@ConditionalOnProperty(prefix = "host.info", name = "mq-mode", havingValue = "1")
public class MqServiceImpl implements MqService {

    private boolean mqConnected;

    @Resource
    private DefaultMQProducer mqProducer;

    @Resource
    private ServerProperties serverProperties;

    @Override
    public boolean getMqStatus() {
        return mqConnected;
    }

    @Override
    public void setMqStatus(boolean isMqOk) {
        mqConnected = isMqOk;
    }

    @Override
    public boolean sendSyncMessage(String message) {

        if (!mqConnected) {
            return false;
        }

        try {
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message(serverProperties.getMqTopic(),
                    serverProperties.getMqTag(),
                    "key_guass",
                    message.getBytes(RemotingHelper.DEFAULT_CHARSET)
            );
            SendResult sendResult = mqProducer.send(msg);

            return  sendResult.getSendStatus() == SendStatus.SEND_OK;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return false;
    }

    @Override
    public void sendAsyncMessage(String message) {

        if (!mqConnected) {
            return;
        }

        try {
            Message msg = new Message(serverProperties.getMqTopic(),
                    serverProperties.getMqTag(),
                    "key_guass",
                    message.getBytes(RemotingHelper.DEFAULT_CHARSET));
            mqProducer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("<======= guass send ok ===>> ");
                }

                @Override
                public void onException(Throwable throwable) {

                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}